package com.googlecode.excavator.pref;

import java.text.DecimalFormat;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.Scanner;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;

import org.apache.log4j.Logger;
import org.springframework.context.support.ClassPathXmlApplicationContext;

import com.googlecode.excavator.pref.service.SayHelloService;

public class ConsumerPerf implements PerfReporter {

	private final Logger logger = Logger.getLogger(ConsumerPerf.class);
	private final SayHelloService sayHelloService;
	private final ExecutorService executor = Executors.newCachedThreadPool(new ThreadFactory() {
		
		@Override
		public Thread newThread(Runnable arg0) {
			Thread thread  =new Thread(arg0);
			thread.setDaemon(true);
			return thread;
		}
	});
	private final List<PerfRunner> perfRunners = new ArrayList<PerfRunner>();
	private int concurrent = 10;
	private boolean isStarted = false;
	
	
	public ConsumerPerf() {
		ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("classpath:/spring-consumer.xml");
		context.registerShutdownHook();
		sayHelloService = (SayHelloService)context.getBean("sayHelloServiceG1");
		
		logger.info("perf started...");
		thread.setDaemon(true);
		thread.start();
		
		Scanner scanner = new Scanner(System.in);
		while( scanner.hasNextLine() ) {

			final String cmd = scanner.nextLine();
			if( cmd.startsWith("c=") ) {
				concurrent = Integer.valueOf(cmd.split("=")[1]);
				logger.info("concurrent changeTo "+concurrent);
				start();
			}
			else if( cmd.startsWith("start") ) {
				start();
			}
			else if( cmd.startsWith("stop") ) {
				stop();
			}
			else if( cmd.equals("quit") ) {
				stop();
				executor.shutdown();
				break;
			}
			else if( !cmd.isEmpty() ) {
				logger.info("unknow command: "+cmd);
			}
			else {
				//
			}
		}
		
		logger.info("perf stoped. bye!");
		
	}
	
	private void start() {
		isStarted = true;
		int diff = perfRunners.size() - concurrent;
		if( diff > 0 ) {
			//kill
			int counter = diff;
			Iterator<PerfRunner> it = perfRunners.iterator();
			while( it.hasNext() && counter-- > 0 ) {
				it.next().kill();
				it.remove();
			}
		} else {
			//new
			int counter = -diff;
			for( int i=0; i< counter; i++ ) {
				PerfRunner perfRunner = new PerfRunner(new Perf(){

					@Override
					public boolean perf() {
						try {
							sayHelloService.sayTo("dukun");
							return true;
						}catch(Throwable t) {
							return false;
						}
					}
					
				}, this);
				perfRunners.add(perfRunner);
				executor.execute(perfRunner);
			}
		}
		
		logger.info("start... concurrent="+perfRunners.size());
		
	}
	
	private void stop() {
		isStarted = false;
		Iterator<PerfRunner> it = perfRunners.iterator();
		while( it.hasNext() ) {
			it.next().kill();
			it.remove();
		}
		logger.info("stop...");
	}
	
	private AtomicLong costs = new AtomicLong(0);
	private AtomicInteger counters = new AtomicInteger(0);
	private AtomicInteger fails = new AtomicInteger(0);
	
	@Override
	public void report(long cost, boolean successed) {
		costs.addAndGet(cost);
		counters.incrementAndGet();
		if( !successed ) {
			fails.incrementAndGet();
		}
	}
	
	private Thread thread = new Thread() {
		public void run() {
			final int sleep = 5000;
			final DecimalFormat formater = new DecimalFormat("#0.##");
			for(;;) {
				try {
					long start = System.currentTimeMillis();
					Thread.sleep(sleep);
					long diff = System.currentTimeMillis() - start;
					if( isStarted ) {
						long total = counters.longValue();
						double rt = costs.longValue()*1.0/total;
						double failRate = fails.longValue()*1.0/total;
						long qps = total * 1000 / diff;
						String log = String.format("concurrent=%s; qps=%s; rt=%sms; fail_rate=%s;", 
								concurrent, qps, formater.format(rt), formater.format(failRate));
						logger.info(log);
						
						//clean
						costs.set(0);
						counters.set(0);
						fails.set(0);
						
					}
				}catch(Exception e) {
					//
				}
			}
		}
	};
	
	public static void main(String... args) {
		new ConsumerPerf();
		System.exit(0);
	}
	
}
